package org.jetlinks.pro.things.impl;

import lombok.AllArgsConstructor;
import org.jetlinks.core.message.ThingMessage;
import org.jetlinks.core.message.function.ThingFunctionInvokeMessage;
import org.jetlinks.core.things.*;
import org.jetlinks.pro.things.ThingsConnectorManager;
import reactor.core.publisher.Flux;

import java.util.function.Function;


/**
 *  RPC支持链，用于对RPC请求做处理
 *
 * @author zhouhao
 * @since 1.12
 */
@AllArgsConstructor
public class DefaultRpcRpcSupportChain implements ThingRpcSupportChain {

    //物连接器管理器,用于管理连接器,并从连接器中获取数据进行持久化等处理
    private final ThingsConnectorManager connectorManager;

    /**
     * RPC调用
     *
     * @param message RPC请求消息
     * @return 多个RPC的调用结果
     */
    @Override
    public Flux<? extends ThingMessage> call(ThingMessage message, ThingRpcSupport next) {
        MetadataId metadataId = getMetadataId(message);
        if (metadataId == null) {
            return next.call(message);
        }
        ThingId thingId = ThingId.of(message.getThingType(), message.getThingId());

        //获取连接器，如果支持rpc则使用rpc进行调用
        return connectorManager
            .getConnector(thingId, metadataId)
            .filter(connector -> connector.isWrapperFor(ThingRpcSupport.class))
            .map(connector -> connector
                .unwrap(ThingRpcSupport.class)
                .call(message)
                .cast(ThingMessage.class)
            )
            .defaultIfEmpty(Flux.defer(() -> next.call(message).cast(ThingMessage.class)))
            .flatMapMany(Function.identity());
    }

    /**
     * 获取物模型中的功能ID
     *
     * @param message 物消息
     * @return 功能ID
     */
    private MetadataId getMetadataId(ThingMessage message) {
        if (message instanceof ThingFunctionInvokeMessage) {
            return MetadataId.of(ThingMetadataType.function, ((ThingFunctionInvokeMessage<?>) message).getFunctionId());
        }
        return null;
    }

}
